package com.learn.rabbit.base.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import javax.annotation.Resource;

/**
 * Rabbitmq配置类
 *
 * @author zhousiqi
 * @version 1.0.0
 * @date 2019/2/26 16:35
 */
@Configuration
@Slf4j
public class RabbitmqConfig {

    @Resource
    private Environment env;

    @Resource
    private CachingConnectionFactory connectionFactory;

    @Resource
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Resource
    private QueueConfig queueConfig;

    @Resource
    private ExchangeConfig exchangeConfig;

    /**
     * routing_key
     */
    public static final String DIRECT_ROUTING_KEY_ONE = "info.orange.base";
    /**
     * routing_key
     */
    public static final String DIRECT_ROUTING_KEY_TWO = "info.rabbit.base";

    /**
     * routing_key
     */
    public static final String TOPIC_ROUTING_KEY_ONE = "info.orange.dev";
    /**
     * routing_key
     */
    public static final String TOPIC_ROUTING_KEY_TWO = "info.rabbit.dev";


    /**
     * binding_key
     */
    public static final String DIRECT_BINDING_KEY_ONE = DIRECT_ROUTING_KEY_ONE;
    /**
     * binding_key
     */
    public static final String DIRECT_BINDING_KEY_TWO = DIRECT_ROUTING_KEY_TWO;

    /**
     * binding_key
     */
    public static final String TOPIC_BINDING_KEY_ONE = "*.orange.*";
    /**
     * binding_key
     */
    public static final String TOPIC_BINDING_KEY_TWO = "info.*.*";

    /**
     * singleListenerContainer
     */
    public static final String SINGLE_LISTENER_CONTAINER = "singleListenerContainer";
    /**
     * multiListenerContainer
     */
    public static final String MULTI_LISTENER_CONTAINER = "multiListenerContainer";


    /**
     * 单一消费者
     */
    @Bean(name = SINGLE_LISTENER_CONTAINER)
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return factory;
    }

    /**
     * 多个消费者
     */
    @Bean(name = MULTI_LISTENER_CONTAINER)
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.concurrency", int.class));
        factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.max-concurrency", int.class));
        factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.prefetch", int.class));
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(
                (correlationData, ack, cause) ->
                        log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause)
        );
        rabbitTemplate.setReturnCallback(
                (message, replyCode, replyText, exchange, routingKey) ->
                        log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message)
        );
        return rabbitTemplate;
    }

    /**
     * 将消息队列1和direct交换机进行绑定
     */
    @Bean
    public Binding bindingOne() {
        return BindingBuilder
                // 队列
                .bind(queueConfig.firstQueue())
                // 交换机
                .to(exchangeConfig.directExchange())
                // binding key
                .with(DIRECT_BINDING_KEY_ONE);
    }

    /**
     * 将消息队列2和direct交换机进行绑定
     */
    @Bean
    public Binding bindingTwo() {
        return BindingBuilder
                .bind(queueConfig.secondQueue())
                .to(exchangeConfig.directExchange())
                // binding key
                .with(DIRECT_BINDING_KEY_TWO);
    }

    /**
     * 将消息队列2和fanout交换机进行绑定
     */
    @Bean
    public Binding bindingOneWithFanout() {
        return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.fanoutExchange());
    }

    /**
     * 将消息队列2和fanout交换机进行绑定
     */
    @Bean
    public Binding bindingOneWithFanout2() {
        return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.fanoutExchange());
    }

    /**
     * 将消息队列1和topic交换机进行绑定
     */
    @Bean
    public Binding bindingOneWithTopic() {
        return BindingBuilder
                .bind(queueConfig.firstQueue())
                .to(exchangeConfig.topicExchange())
                // binding key(这里写匹配规则)
                .with(TOPIC_BINDING_KEY_ONE);
    }

    /**
     * 将消息队列2和topic交换机进行绑定
     */
    @Bean
    public Binding bindingOneWithTopic2() {
        return BindingBuilder
                .bind(queueConfig.secondQueue())
                .to(exchangeConfig.topicExchange())
                // binding key(这里写匹配规则)
                .with(TOPIC_BINDING_KEY_TWO);
    }
}
